/*
 * Copyright © 2017 eqxiu.com 北京中网易企秀科技有限公司  All rights reserved.
 */

package cn.hermit.nio.server;

import cn.hermit.core.Disposable;
import cn.hermit.nio.core.ChangeRequest;
import cn.hermit.nio.core.EchoRequestHandler;
import cn.hermit.nio.core.RequestHandler;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * @author JackGao 高建国
 * @Reference http://rox-xmlrpc.sourceforge.net/niotut/
 * @description //TODO 设计说明
 * @date 17/5/2017
 */
public class NioServer implements Runnable, Disposable {

    private static final Logger logger = Logger.getLogger(NioServer.class.getName());

    private volatile boolean closed;

    // The host:port combination to listen on
    private InetAddress hostAddress;
    private int port = 9000;

    // The channel on which we'll accept connections
    private ServerSocketChannel serverChannel;

    // The selector we'll be monitoring
    private final Selector selector;

    // The buffer into which we'll read data when it's available
    private ByteBuffer readBuffer = ByteBuffer.allocate(8192);

//    private final NioWorker[] workerArr;

    private final ThreadPoolExecutor poolExecutor;

    public NioServer(int port) throws IOException {
        this(null, port, 0);
    }

    public NioServer(InetAddress hostAddress, int port) throws IOException {
        this(hostAddress, port, 0);
    }

    public NioServer(InetAddress hostAddress, int port, int workerCount) throws IOException {
        if (workerCount <= 0) {
            workerCount = Runtime.getRuntime().availableProcessors();
        }
        this.hostAddress = hostAddress;
        if(port > 9000) {
            this.port = port;
        }
        this.selector = this.initSelector();
//        workerArr = new NioWorker[workerCount];
        poolExecutor = new ThreadPoolExecutor(2, workerCount, 1000, TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(workerCount));
//        for (int i = 0; i < workerCount; ++i) {
//            NioWorker worker = new NioWorker();
//            workerArr[i] = worker;
//            Thread t = new Thread(worker, "Worker-" + i);
//            t.start();
//            logger.log(Level.INFO, "Nio {0}[{1}] started", new String[]{t.getName(), String.valueOf(t.getId())});
//        }
        new Thread(this, "Server").start();
        logger.log(Level.INFO, "Nio Server started at @{0}:{1}", new String[]{hostAddress == null ? "0.0.0.0" : hostAddress.getHostName(), String.valueOf(port)});
    }

    public void run() {
        while (!closed) {
            try {

                // Process any pending changes
                synchronized (this.changeRequests) {
                    Iterator changes = this.changeRequests.iterator();
                    while (changes.hasNext()) {
                        ChangeRequest change = (ChangeRequest) changes.next();
                        switch (change.type) {
                            case ChangeRequest.CHANGEOPS:
                                SelectionKey key = change.socket.keyFor(this.selector);
                                if (key != null && key.isValid()) {
                                    key.interestOps(change.ops);
                                }
                        }
                    }
                    this.changeRequests.clear();
                }

                // Wait for an event one of the registered channels
                this.selector.select();

                // Iterate over the setHeader of keys for which events are available
                Iterator selectedKeys = this.selector.selectedKeys().iterator();
                while (selectedKeys.hasNext()) {
                    SelectionKey key = (SelectionKey) selectedKeys.next();
                    selectedKeys.remove();

                    if (!key.isValid()) {
                        continue;
                    }

                    // Check what event is available and deal with it
                    if (key.isAcceptable()) {
                        this.accept(key);
                    } else if (key.isReadable()) {
                        this.read(key);
                    } else if (key.isWritable()) {
                        this.write(key);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
                logger.log(Level.SEVERE, e.getMessage());
            }
        }
    }

    private void accept(SelectionKey key) throws IOException {
        // For an accept to be pending the channel must be a server socket channel.
        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();

        // Accept the connection and make it non-blocking
        SocketChannel socketChannel = serverSocketChannel.accept();
        Socket socket = socketChannel.socket();
        socketChannel.configureBlocking(false);

        // Register the new SocketChannel with our Selector, indicating
        // we'd like to be notified when there's data waiting to be read
        socketChannel.register(this.selector, SelectionKey.OP_READ);
    }

    private void read(SelectionKey key) throws IOException {
        final SocketChannel socketChannel = (SocketChannel) key.channel();

        // Clear out our read buffer so it's ready for new data
        this.readBuffer.clear();

        // Attempt to read off the channel
        int numRead;
        try {
            numRead = socketChannel.read(this.readBuffer);
        } catch (IOException e) {
            // The remote forcibly closed the connection, cancel
            // the selection key and close the channel.
            key.cancel();
            socketChannel.close();
            return;
        }

        if (numRead == -1) {
            // Remote entity shut the socket down cleanly. Do the
            // same from our end and cancel the channel.
            key.channel().close();
            key.cancel();
            return;
        }
//        int workerIndex = socketChannel.hashCode() % workerArr.length;
        // Hand the data off to our worker thread
//        this.workerArr[workerIndex].processData(this, socketChannel, this.readBuffer.array(), numRead);
        final int count = numRead;
        final byte[] data = this.readBuffer.array();
        poolExecutor.execute(new Runnable() {
            @Override
            public void run() {
                byte[] dataCopy = new byte[count];
                System.arraycopy(data, 0, dataCopy, 0, count);
                getRequestHandler().handle(new ServerDataEvent(NioServer.this, socketChannel, dataCopy));
            }
        });
    }

    private void write(SelectionKey key) throws IOException {
        SocketChannel socketChannel = (SocketChannel) key.channel();

        synchronized (this.pendingData) {
            List queue = (List) this.pendingData.get(socketChannel);

            // Write until there's not more data ...
            while (!queue.isEmpty()) {
                ByteBuffer buf = (ByteBuffer) queue.get(0);
                socketChannel.write(buf);
                if (buf.remaining() > 0) {
                    // ... or the socket's buffer fills up
                    break;
                }
                queue.remove(0);
            }

            if (queue.isEmpty()) {
                // We wrote away all data, so we're no longer interested
                // in writing on this socket. Switch back to waiting for
                // data.
                key.interestOps(SelectionKey.OP_READ);
            }
        }
    }

    private Selector initSelector() throws IOException {
        // Create a new selector
        Selector socketSelector = SelectorProvider.provider().openSelector();

        // Create a new non-blocking server socket channel
        this.serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);

        // Bind the server socket to the specified address and port
        InetSocketAddress isa = new InetSocketAddress(this.hostAddress, this.port);
        serverChannel.socket().bind(isa);

        // Register the server socket channel, indicating an interest in
        // accepting new connections
        serverChannel.register(socketSelector, SelectionKey.OP_ACCEPT);

        return socketSelector;
    }

    // A list of ChangeRequest instances
    private List changeRequests = new LinkedList();

    // Maps a SocketChannel to a list of ByteBuffer instances
    private Map pendingData = new HashMap();

    public void send(SocketChannel socket, byte[] data) {
        synchronized (this.changeRequests) {
            // Indicate we want the interest ops setHeader changed
            this.changeRequests.add(new ChangeRequest(socket, ChangeRequest.CHANGEOPS, SelectionKey.OP_WRITE));

            // And queue the data we want written
            synchronized (this.pendingData) {
                List queue = (List) this.pendingData.get(socket);
                if (queue == null) {
                    queue = new ArrayList();
                    this.pendingData.put(socket, queue);
                }
                queue.add(ByteBuffer.wrap(data));
            }
        }

        // Finally, wake up our selecting thread so it can make the required changes
        this.selector.wakeup();
    }

    @Override
    public void destroy() {
        closed = true;
        try {
            closed = true;
            this.selector.close();
            this.changeRequests.clear();
            this.pendingData.clear();
            this.serverChannel.close();
//            for(NioWorker worker : this.workerArr) {
//                worker.destroy();
//            }
            poolExecutor.shutdown();
            while (!poolExecutor.isTerminated()) {

            }
            this.readBuffer.clear();
        } catch (IOException e) {
            //ignored
        }

    }

    public RequestHandler getRequestHandler() {
        return new EchoRequestHandler();
    }

    public static void main(String[] args) {
        try {
            new NioServer(9090);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}